package com.inspur.cloud.service.dataspace;

import com.inspur.cloud.configuration.AmbariConfig;
import com.inspur.cloud.configuration.ApiConfig;
import com.inspur.cloud.entity.dataspace.dto.HdfsInfo;
import com.inspur.cloud.exception.HdfsException;
import com.inspur.cloud.util.TConstants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.client.HdfsAdmin;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 对HDFS组件操作。
 */
@Service
public class HdfsService {

    private static Logger log = LoggerFactory.getLogger(HdfsService.class);


    @Autowired
    private ApiConfig apiConfig;
    @Autowired
    private ClusterConfigUtil clusterConfigUtil;
    @Autowired
    private AmbariConfig ambariConfig;

    /**
     * 获取hdfs文件系统实例
     * @return
     */
    public  FileSystem getFileSystem(){
        FileSystem fs;
        try{
            Map<String, String> map = clusterConfigUtil.getHdfsConfig();
            log.debug("获取的配置信息:" + map);
            String uri = map.get("fs.defaultFS");
            if (uri == null || "null".equals(uri) || "".equals(uri)) {
                log.error("标识为" + TConstants.CLUSTER_NAME + "的集群，在大数据集群配置中配置信息错误！");
                throw new HdfsException("标识为" + TConstants.CLUSTER_NAME
                        + "的集群，在大数据集群配置中配置信息错误！");
            }
            //加载HDFS默认配置文件，core-default.xml和hdfs-default.xml
            Configuration conf = new Configuration();
            String user = "";
            for (Map.Entry<String, String> entry : map.entrySet()) {
                if ("fs.defaultFS".equals(entry.getKey())) {
                    uri = entry.getValue();
                } else if ("hdfs_user".equals(entry.getKey())) {
                    user = entry.getValue();
                } else if((!"hdfs_user_keytab".equals(entry.getKey()))&&(!"hdfs_user".equals(entry.getKey()))&&(!"hdfs_principal_name".equals(entry.getKey()))) {
                    conf.set(entry.getKey(), entry.getValue());
                }
            }
            //如果没有这个属性，默认获取的是本地文件系统
            conf.set("fs.defaultFS",uri);
            conf.set("dfs.cluster.administrators", user);
            String authorizationType = map.get("hadoop.security.authentication");
            String isKerberos = map.get("hadoop.security.authorization");
            if ("kerberos".equals(authorizationType) && "true".equals(isKerberos)) {
                //princal账号
                String principal_name;
                //keytab文件路径
                String keypath;
                //kerberos配置文件路径
                String krbConf;
                principal_name = ambariConfig.getDataspacePrincipal();
                if (apiConfig.isDev()) {
                    krbConf = Thread.currentThread().getContextClassLoader().getResource("tempfiles/krb5.conf").toString();
                    krbConf = krbConf.substring(5);
                    //本地调试的keypath
                    keypath = Thread.currentThread().getContextClassLoader().getResource("tempfiles/dataspace.keytab").toString();
                    keypath = keypath.substring(5);
                    log.info("keytab路径是:" + keypath);
                } else {
                    krbConf = TConstants.KRB5CONF_PATH;
                    //服务器上管理员keypath
                    keypath = ambariConfig.getDataspaceKeytab();
                    log.info("keytab路径是:"+keypath);
                }
                log.info("krbConf目录:" + krbConf);
                System.clearProperty("java.security.krb5.conf");
                System.setProperty("java.security.krb5.conf", krbConf);
                //将HDFS配置文件加载进入kerberos中
                UserGroupInformation.setConfiguration(conf);
                try {
                    sun.security.krb5.Config.refresh();
                } catch (Exception e) {
                    log.error("刷新kerberos config信息报错：", e);
                }
                //kerberos登录
                UserGroupInformation.loginUserFromKeytab(principal_name, keypath);
            }
            System.setProperty("HADOOP_USER_NAME","dataspace");
            fs = FileSystem.get(conf);
            log.info("得到文件系统实例[{}]",fs);
        }catch(Exception e){
            log.error("根据集群标识" + TConstants.CLUSTER_NAME + "获取文件系统出错！", e);
            throw new HdfsException("根据集群标识" + TConstants.CLUSTER_NAME + "获取文件系统出错！");
        }
        return fs;
    }
    /**
     * 管理员查询总HDFS基本状态信息
     * @return capacity容量 remaining 使用情况 used还剩多少(字节)
     */
    public HdfsInfo getHDFSStatus(){
        HdfsInfo hdfsInfo = new HdfsInfo();
        FsStatus fss;
        try(FileSystem fs = getFileSystem()) {
            fss = fs.getStatus();
            log.info("HDFS基本状态信息[{}]实例",fss);
        } catch (IOException e) {
            log.error("查询失败",e);
            throw new HdfsException("查询HDFS基本状态信息失败"+e.getMessage());
        }
        hdfsInfo.setCapacity(fss.getCapacity());
        hdfsInfo.setRemaining(fss.getRemaining());
        hdfsInfo.setUsed(fss.getUsed());
        return hdfsInfo;
    }
    /**
     * 租户用户查看自己HDFS的空间情况
     * @param path 目录名
     * @return
     * @throws HdfsException
     */
    public HdfsInfo getContentSummary(String path) {
        HdfsInfo hdfsInfo = new HdfsInfo();
        try (FileSystem fs = getFileSystem()) {
            if (!fs.exists(new Path(path))) {
                log.warn("目录[{}]不存在", path);
                throw new HdfsException("目录不存在!");
            } else {
                ContentSummary contentSummary = fs.getContentSummary(new Path(path));
                //得到该目录的总空间大小
                long capacity = contentSummary.getSpaceQuota();
                //得到该目录的使用大小
                long used =contentSummary.getSpaceConsumed();
                //得到该目录的剩余大小
                long remaining = capacity - used;
                //得到该目录下现有的文件和文件夹数
                long fileAndDirectoryCount = contentSummary.getFileAndDirectoryCount();
                //现有的文件数
                long fileCount = contentSummary.getFileCount();
                //现有的目录数
                long directoryCount = contentSummary.getDirectoryCount();
                //得到该目录下配置的文件总数
                long nameQuota = contentSummary.getQuota();
                hdfsInfo.setCapacity(capacity);
                hdfsInfo.setRemaining(remaining);
                hdfsInfo.setUsed(used);
                hdfsInfo.setFileAndDirectoryCount(fileAndDirectoryCount);
                hdfsInfo.setNameQuota(nameQuota);
                hdfsInfo.setFileCount(fileCount);
                hdfsInfo.setDirectoryCount(directoryCount);
                return hdfsInfo;
            }
        } catch(Exception e) {
            log.error("***getContentSummary***Error***Message", e);
            throw new HdfsException("getContentSummary异常!" + e.getMessage());
        }
//        return hdfsInfo;
    }
    /**
     * 给新建的hdfs目录设置所属用户和用户组
     * @param dir 目录名
     * @param username 设置所属用户
     * @param groupname 设置所属用户组
     * @return
     */
    public  void setOwner(String dir,String username,String groupname) {
        Path hdfsDir = new Path(dir);
        try(FileSystem fs = getFileSystem()) {
            if (!fs.exists(new Path(dir))) {
                log.warn("该目录[{}]不存在,请输入正确的路径", dir);
//                throw new HdfsException("目录不存在，请输入正确路径");
            } else {
                if(TConstants.ADMIN_NAME.equals(username)){
                    fs.setOwner(hdfsDir, TConstants.DATASPACE_USER, TConstants.DATASPACE_USER);
                    log.info("设置文件目录所属用户和用户组成功,user=[{}],group=[{}]", TConstants.DATASPACE_USER,TConstants.DATASPACE_USER);
                }else {
                    fs.setOwner(hdfsDir,username,groupname);
                }
            }
        } catch (IOException e) {
            log.error("创建文件目录出现异常!", e);
            throw new HdfsException("给新建的hdfs目录设置所属用户和用户组失败!" + e.getMessage());
        }
    }

    /**
     * 展示一个HDFS目录下的子目录(全路径显示)
     * @param dir
     * @throws IOException
     */
    public  List<String> listDir(String dir) {
        List<String> list = new ArrayList<>();
        FileStatus[] files;
        try(FileSystem fs = getFileSystem()) {
            if (!fs.exists(new Path(dir))) {
                log.warn("该目录[{}]不存在,请输入正确的路径", dir);
//                throw new HdfsException("目录不存在，请输入正确路径");
            } else {
                files = fs.listStatus(new Path(dir));
                for (FileStatus file : files) {
                    if(fs.isDirectory(file.getPath())){
                        list.add(file.getPath().toString());
                    }
                }
                log.info("目录[{}]下的子目录有[{}]",dir,list);
            }
        } catch (IOException e) {
            log.error("查询目录下的文件夹和文件失败!",e);
            throw new HdfsException("查询目录下的文件夹和文件失败"+e.getMessage());
        }
        return list;
    }

    /**
     * 创建HDFS目录。
     * @param dir
     * @return 成功或失败
     */
    public void makeDir(String dir) {
        Path hdfsDir = new Path(dir);
        try(FileSystem fs= getFileSystem()){
            if (fs.exists(hdfsDir)) {
                log.warn("该文件目录已经存在,path=[{}]", dir);
//                throw new HdfsException("创建目录失败:该文件目录已经存在");
            } else {
                //不存在则创建
                FsPermission fp = new FsPermission(FsAction.ALL,FsAction.NONE,FsAction.NONE);
                fs.mkdirs(hdfsDir,fp);
                log.info("创建文件目录成功,path=[{}]", dir);
            }
        }catch (Exception e){
            log.error("创建文件目录失败,path=[{}]", dir, e);
            throw new HdfsException("HDFS创建文件目录失败："+e.getMessage());
        }
    }
    public void setPermission(String dir) {
        Path hdfsDir = new Path(dir);
        try(FileSystem fs= getFileSystem()){
            if (fs.exists(hdfsDir)) {
                FsPermission fp = new FsPermission(FsAction.ALL,FsAction.ALL,FsAction.ALL);
                fs.setPermission(hdfsDir,fp);
                log.info("修改目录权限成功,path=[{}]", dir);
            }
        }catch (Exception e){
            log.error("修改目录权限失败,path=[{}]", dir, e);
            throw new HdfsException("HDFS修改目录权限失败："+e.getMessage());
        }
    }


    /**
     * 设置HDFS指定目录下的文件数和配额
     */
    public void setQuota(String dir, long capacity, int fileNum) {
        HdfsAdmin hdfsAdmin ;
        Path hdfsDir = new Path(dir);
        try (FileSystem fs = getFileSystem()) {
            hdfsAdmin = new HdfsAdmin(fs.getUri(), fs.getConf());
            if(fs.exists(hdfsDir)){
                hdfsAdmin.setQuota(hdfsDir, fileNum);
                log.info("成功设置HDFS的[{}]目录的文件数为[{}]", dir, fileNum);
                hdfsAdmin.setSpaceQuota(hdfsDir, capacity);
                log.info("成功设置HDFS的[{}]目录的空间大小为{}", dir, capacity);
            }else {
                log.warn("该[{}]HDFS目录不存在", dir);
//                throw new HdfsException("设置目录文件数失败:目录不存在");
            }
        } catch (IOException e) {
            log.error("设置目录文件数和配额失败,path=[{}]", dir, e);
            throw new HdfsException("HDFS指定目录文件总数失败："+e.getMessage());
        }
    }

    /**
     * 移动数据到垃圾站,或者从垃圾站恢复。
     * @param dir
     * @param newDir
     * @return
     */
    public  void mvToCollection(String dir,String newDir){
        try(FileSystem fs = getFileSystem()){
            if (!fs.exists(new Path(dir))) {
                log.error("目录[{}]不存在，无法移动", dir);
//                    throw new HdfsException("移动HDFS上目录"+dir+"失败！");
            }
            if (fs.exists(new Path(newDir))) {
                log.error("新目录[{}]已经存在，无法移动", newDir);
//                    throw new HdfsException("移动HDFS上目录"+dir+"失败！");
            }
            if (!fs.exists(new Path(newDir)) && fs.exists(new Path(dir))) {
                fs.rename(new Path((dir)), new Path(newDir));
                log.info("移动成功:[{}]移动到[{}]", dir, newDir);
            }
        }catch (Exception e){
            log.error("移动HDFS上目录:[{}] 失败！", dir, e);
            throw new HdfsException("移动HDFS上目录"+dir+"失败！");
        }
    }
    /**
     * 租户用户删除指定目录
     * @param dir
     * @return
     */
    public  void removeDir(String dir){
        try(FileSystem fs = getFileSystem()) {
            //删除一个目录，true表示递归删除
            if(fs.exists(new Path(dir))){
                fs.delete(new Path(dir),true);
                log.info("删除目录[{}]成功",dir);
            }
            else{
                log.warn("目录[{}]不存在,无法删除",dir);
//                throw new HdfsException("目录[{}]不存在,无法删除");
            }
        } catch (IOException e) {
            log.error("删除目录[{}]失败",dir,e);
            throw new HdfsException("删除目录"+dir+"失败！");
        }
    }
}
